#include "in.h"

size_t CalcTopicDataUnitSize()
{
	return sizeof(uint64_t) + sizeof(uint64_t);
}

uint64_t MemWrite_TopicDataUnit( char *unit_offset_ptr , uint64_t topic_offset , uint64_t msg_offset )
{
	char		*p = unit_offset_ptr ;
	uint64_t	pack_len ;
	
	*(uint64_t*)p = hton64(topic_offset) ;
	p += sizeof(uint64_t) ;
	
	*(uint64_t*)p = hton64(msg_offset) ;
	p += sizeof(uint64_t) ;
	
	*(uint64_t*)p = hton64(0) ;
	
	pack_len = p - unit_offset_ptr ;
	
	return pack_len;
}

uint64_t MemRead_TopicDataUnit( char *unit_offset_ptr , uint64_t *p_topic_offset , uint64_t *p_msg_offset )
{
	char		*p = unit_offset_ptr ;
	uint64_t	topic_offset ;
	uint64_t	msg_offset ;
	uint64_t	unpack_len ;
	
	topic_offset = ntoh64(*(uint64_t*)p) ;
	if( p_topic_offset )
		(*p_topic_offset) = topic_offset ;
	if( topic_offset == 0 || topic_offset == UINT64_MAX )
		return topic_offset;
	p += sizeof(uint64_t) ;
	
	msg_offset = ntoh64(*(uint64_t*)p) ;
	if( p_msg_offset )
		(*p_msg_offset) = msg_offset ;
	p += sizeof(uint64_t) ;
	
	unpack_len = p - unit_offset_ptr ;
	
	return unpack_len;
}

int CreateTopicDataFile( struct AppendBuffer *comm_data , struct MQBroker *broker , struct MQStore *store , struct MQTopic *topic , size_t begin_offset , struct MQFileMapping **pp_topic_file , size_t *p_file_size )
{
	struct MQFileMapping	*topic_file = NULL ;
	int			file_fd ;
	off_t			seek_file_offset ;
	ssize_t			write_file_len ;
	
	int			nret = 0 ;
	
	topic_file = (struct MQFileMapping *)malloc( sizeof(struct MQFileMapping) ) ;
	if( topic_file == NULL )
	{
		ERRORLOG( "malloc failed , errno[%d]" , errno )
		if( comm_data )
			MEMPACK_RESULT( return -11 , MQSDK_ERROR_MALLOC , comm_data )
		return -1;
	}
	memset( topic_file , 0x00 , sizeof(struct MQFileMapping) );
	
	memset( topic_file->path_file_name , 0x00 , sizeof(topic_file->path_file_name) );
	snprintf( topic_file->path_file_name , sizeof(topic_file->path_file_name)-1 , "%s/%020zu.cmqt" , topic->topic_path , begin_offset );
	file_fd = open( topic_file->path_file_name , O_CREAT|O_EXCL|O_RDWR|O_TRUNC , 00777 ) ;
	if( file_fd == -1 )
	{
		ERRORLOG( "open[%s] failed , errno[%d]" , topic_file->path_file_name , errno )
		free( topic_file );
		if( comm_data )
			MEMPACK_RESULT( return -12 , MQSDK_ERROR_CREATE_TOPIC_FILE , comm_data )
		return -2;
	}
	else
	{
		DEBUGLOG( "open[%s] ok" , topic_file->path_file_name )
	}
	
	topic_file->begin_offset = begin_offset ;
	topic_file->file_size = broker->topic_file_size ;
	
	seek_file_offset = lseek( file_fd , topic_file->file_size-1 , SEEK_SET ) ;
	if( seek_file_offset == -1 )
	{
		ERRORLOG( "lseek[%s] offset[%zu] failed , errno[%d]" , topic_file->path_file_name , topic_file->file_size-1 , errno )
		close( file_fd );
		unlink( topic_file->path_file_name );
		free( topic_file );
		if( comm_data )
			MEMPACK_RESULT( return -13 , MQSDK_ERROR_LSEEK_MSG_DATA_FILE , comm_data )
		return -3;
	}
	else
	{
		DEBUGLOG( "lseek[%s] offset[%zu] ok" , topic_file->path_file_name , topic_file->file_size-1 )
	}
	
	write_file_len = write( file_fd , "\0" , 1 ) ;
	if( write_file_len == -1 )
	{
		ERRORLOG( "write[%s] failed , errno[%d]" , topic_file->path_file_name , errno )
		close( file_fd );
		unlink( topic_file->path_file_name );
		free( topic_file );
		if( comm_data )
			MEMPACK_RESULT( return -14 , MQSDK_ERROR_LWRITE_TOPIC_FILE , comm_data )
		return -4;
	}
	else
	{
		DEBUGLOG( "write[%s] ok" , topic_file->path_file_name )
	}
	
	topic_file->mmap_size = topic_file->file_size ;
	topic_file->file_mmap = mmap( NULL , topic_file->file_size , PROT_READ|PROT_WRITE , MAP_SHARED , file_fd , 0 ) ;
	if( topic_file->file_mmap == (void*)(-1) )
	{
		ERRORLOG( "mmap[%s] failed , errno[%d]" , topic_file->path_file_name , errno )
		close( file_fd );
		unlink( topic_file->path_file_name );
		free( topic_file );
		if( comm_data )
			MEMPACK_RESULT( return -15 , MQSDK_ERROR_LWRITE_MSG_DATA_FILE , comm_data )
		return -5;
	}
	else
	{
		DEBUGLOG( "mmap[%s] size[%zu] ok , addr[0x%X]" , topic_file->path_file_name , topic_file->file_size , topic_file->file_mmap )
	}
	
	close( file_fd );
	
	memset( topic_file->file_mmap , 0x00 , topic_file->file_size );
	topic_file->fill_ptr = topic_file->file_mmap ;
	topic_file->over_end_ptr = topic_file->file_mmap + topic_file->file_size ;
	DEBUGLOG( "topic_file->over_end_ptr[0x%X]" , topic_file->over_end_ptr )
	
	DEBUGLOG( "write topic file head" )
	memcpy( topic_file->fill_ptr+MQ_TOPIC_FILE_HEADER_FORMAT_MAGIC_OFFSET , MQ_TOPIC_FILE_HEADER_FORMAT_MAGIC_VALUE , MQ_TOPIC_FILE_HEADER_FORMAT_MAGIC_LENGTH );
	*(topic_file->fill_ptr+MQ_TOPIC_FILE_HEADER_FORMAT_VERSION_OFFSET) = MQ_TOPIC_FILE_RESTORE_FORMAT_VERSION_1 ;
	*((uint64_t*)(topic_file->fill_ptr+MQ_TOPIC_FILE_HEADER_FORMAT_MSG_OFFSET_OFFSET)) = hton64(begin_offset) ;
	*((uint64_t*)(topic_file->fill_ptr+MQ_TOPIC_FILE_HEADER_FORMAT_ORI_FILE_SIZE_OFFSET)) = hton64(topic_file->file_size) ;
	*(topic_file->fill_ptr+MQ_TOPIC_FILE_HEADER_FORMAT_IS_CLEAN_OFFSET) = MQ_TOPIC_FILE_HEADER_FORMAT_NOT_CLEAN ;
	topic_file->fill_ptr += MQ_TOPIC_FILE_HEAD_SIZE ;
	DEBUGLOG( "topic_file->fill_ptr[0x%X]" , topic_file->fill_ptr )
	
	topic->max_topic_offset += MQ_TOPIC_FILE_HEAD_SIZE ;
	DEBUGLOG( "topic->max_topic_offset[%zu]" , topic->max_topic_offset )
	
	INIT_LIST_HEAD( & (topic_file->subscribe_session_list) );
	
	DEBUGLOG( "add topic_file[%s] to topic's topic_file_list" , topic_file->path_file_name )
	list_add_tail( & (topic_file->file_list_node) , & (topic->topic_file_list) );
	topic->topic_file_list_nodes_count++;
	
	if( pp_topic_file )
		(*pp_topic_file) = topic_file ;
	if( p_file_size )
		(*p_file_size) = topic_file->file_size ;
	return 0;
}

int LoadTopicDataFile( struct AppendBuffer *comm_data , struct MQBroker *broker , struct MQStore *store , struct MQTopic *topic , size_t begin_offset , char *filename , struct MQFileMapping **pp_topic_file , size_t *p_file_size )
{
	struct MQFileMapping	*topic_file = NULL ;
	int			file_fd ;
	
	int			nret = 0 ;
	
	topic_file = (struct MQFileMapping *)malloc( sizeof(struct MQFileMapping) ) ;
	if( topic_file == NULL )
	{
		ERRORLOG( "malloc failed , errno[%d]" , errno )
		if( comm_data )
			MEMPACK_RESULT( return -11 , MQSDK_ERROR_MALLOC , comm_data )
		return -1;
	}
	memset( topic_file , 0x00 , sizeof(struct MQFileMapping) );
	
	memset( topic_file->path_file_name , 0x00 , sizeof(topic_file->path_file_name) );
	if( filename == NULL )
		snprintf( topic_file->path_file_name , sizeof(topic_file->path_file_name)-1 , "%s/%020zu.cmqt" , topic->topic_path , begin_offset );
	else
		strncpy( topic_file->path_file_name , filename , sizeof(topic_file->path_file_name)-1 );
	file_fd = open( topic_file->path_file_name , O_RDWR , 00777 ) ;
	if( file_fd == -1 )
	{
		DEBUGLOG( "open[%s] not exist" , topic_file->path_file_name )
		free( topic_file );
		if( comm_data )
			MEMPACK_RESULT( return -12 , MQSDK_ERROR_CREATE_MSG_DATA_FILE , comm_data )
		return -2;
	}
	else
	{
		DEBUGLOG( "open[%s] ok" , topic_file->path_file_name )
	}
	
	topic_file->mmap_size = GetFileSize( NULL , file_fd ) ;
	if( topic_file->mmap_size < MQ_TOPIC_FILE_HEAD_SIZE )
	{
		ERRORLOG( "file size[%z] invalid" , topic_file->mmap_size )
		close( file_fd );
		unlink( topic_file->path_file_name );
		free( topic_file );
		if( comm_data )
			MEMPACK_RESULT( return -13 , MQSDK_ERROR_TOPIC_FILE_SIZE_INVALID , comm_data )
		return -3;
	}
	topic_file->file_mmap = mmap( NULL , topic_file->mmap_size , PROT_READ|PROT_WRITE , MAP_SHARED , file_fd , 0 ) ;
	if( topic_file->file_mmap == (void*)(-1) )
	{
		ERRORLOG( "mmap[%s] failed , errno[%d]" , topic_file->path_file_name , errno )
		close( file_fd );
		unlink( topic_file->path_file_name );
		free( topic_file );
		if( comm_data )
			MEMPACK_RESULT( return -13 , MQSDK_ERROR_LWRITE_MSG_DATA_FILE , comm_data )
		return -3;
	}
	else
	{
		if( filename == NULL )
			topic_file->begin_offset = begin_offset ;
		else
			topic_file->begin_offset = *((uint64_t*)(topic_file->file_mmap+5)) ;
		topic_file->file_size = ntoh64(*((uint64_t*)(topic_file->file_mmap+13))) ;
		DEBUGLOG( "mmap[%s] mmap_size[%zu] file_size[%zu] ok , addr[0x%X]" , topic_file->path_file_name , topic_file->mmap_size , topic_file->file_size , topic_file->file_mmap )
	}
	
	if( p_file_size )
	{
		(*p_file_size) = topic_file->file_size ;
		DEBUGLOG( "ret (*p_file_size)[%zu]" , (*p_file_size) )
	}
	
	close( file_fd );
	
	if( *(topic_file->file_mmap+MQ_TOPIC_FILE_HEADER_FORMAT_IS_CLEAN_OFFSET) == MQ_TOPIC_FILE_HEADER_FORMAT_NOT_CLEAN )
	{
		topic_file->over_end_ptr = topic_file->file_mmap + topic_file->file_size ;
		topic_file->fill_ptr = topic_file->file_mmap + MQ_TOPIC_FILE_HEAD_SIZE ;
		
		INIT_LIST_HEAD( & (topic_file->subscribe_session_list) );
		
		if( topic )
		{
			INFOLOG( "add topic[%s] to topic's topic_file_list" , topic_file->path_file_name )
			list_add_tail( & (topic_file->file_list_node) , & (topic->topic_file_list) );
			topic->topic_file_list_nodes_count++;
		}
	}
	else
	{
		DEBUGLOG( "munmap file_mmap[%p]" , topic_file->file_mmap )
		munmap( topic_file->file_mmap , topic_file->mmap_size );
		DEBUGLOG( "free topic[%p]" , topic )
		free( topic_file ); topic_file = NULL ;
	}
	
	if( pp_topic_file )
	{
		(*pp_topic_file) = topic_file ;
		DEBUGLOG( "ret (*pp_topic_file)[%p]" , (*pp_topic_file) )
	}
	
	return 0;
}

int UnloadTopicDataFile( struct MQBroker *broker , struct MQStore *store , struct MQTopic *topic , struct MQFileMapping *topic_file )
{
	munmap( topic_file->file_mmap , topic_file->file_size );
	
	list_del( & (topic_file->file_list_node) );
	topic->topic_file_list_nodes_count--;
	
	free( topic_file );
	
	return 0;
}

int CleanTopicDataFile( struct MQBroker *broker , struct MQStore *store , struct MQTopic *topic , struct MQFileMapping *topic_file )
{
	char		path_file_name[ PATH_MAX ] ;
	
	int		nret = 0 ;
	
	strcpy( path_file_name , topic_file->path_file_name );
	
	DEBUGLOG( "clean topic_file[%s]" , path_file_name )
	*(topic_file->file_mmap+MQ_TOPIC_FILE_HEADER_FORMAT_IS_CLEAN_OFFSET) = MQ_TOPIC_FILE_HEADER_FORMAT_CLEANNED ;
	
	DEBUGLOG( "UnloadTopicDataFile ... store[%s] topic[%s] topic_file[%s]" , store->store_name , topic->topic_name , path_file_name )
	nret = UnloadTopicDataFile( broker , store , topic , topic_file ) ;
	if( nret )
	{
		ERRORLOG( "UnloadTopicDataFile failed[%d]" , nret )
		return nret;
	}
	else
	{
		DEBUGLOG( "UnloadTopicDataFile ok" )
	}
	
	DEBUGLOG( "truncate topic_file[%s]" , path_file_name )
	truncate( path_file_name , MQ_TOPIC_FILE_HEAD_SIZE );
	
	return 0;
}

int LoadAllTopicDataFiles( struct MQBroker *broker , struct MQStore *store , struct MQTopic *topic )
{
	size_t			begin_size ;
	struct MQFileMapping	*topic_file = NULL ;
	size_t			file_size ;
	uint64_t		read_len ;
	
	int			nret = 0 ;
	
	begin_size = 0 ;
	for( ; ; )
	{
		nret = LoadTopicDataFile( NULL , broker , store , topic , begin_size , NULL , & topic_file , & file_size ) ;
		if( nret == -2 )
		{
			break;
		}
		else if( nret )
		{
			ERRORLOG( "LoadMessageTopicDataFile failed" )
			break;
		}
		else
		{
			INFOLOG( "LoadMessageTopicDataFile ok , begin_size[%zu] file_size[%zu]" , begin_size , file_size )
		}
		
		for( ; ; )
		{
			read_len = MemRead_TopicDataUnit( topic_file->fill_ptr , NULL , NULL ) ;
			if( read_len == 0 || read_len == UINT64_MAX )
			{
				store->max_msg_offset = topic_file->fill_ptr - topic_file->file_mmap + topic_file->begin_offset ;
				break;
			}
			else
			{
				topic_file->fill_ptr += read_len ;
				store->max_msg_offset += read_len ;
			}
		}
		
		begin_size += file_size ;
	}
	
	return 0;
}

int LoadAllTopics( struct MQBroker *broker , struct MQStore *store )
{
	DIR			*dp = NULL ;
	struct dirent		*dtp = NULL ;
	char			topic_path_name[ PATH_MAX ] ;
	struct stat		dir_stat ;
	struct MQTopic	*topic = NULL ;
	
	int			nret = 0 ;
	
	dp = opendir( store->store_path ) ;
	if( dp == NULL )
	{
		printf( "ERROR : opendir[%s] failed , errno[%d]\n" , store->store_path , errno );
		return -1;
	}
	
	for( ; ; )
	{
		dtp = readdir( dp ) ;
		if( dtp == NULL )
			break;
		
		if( STRCMP( dtp->d_name , == , "." ) || STRCMP( dtp->d_name , == , ".." ) )
			continue;
		
		memset( topic_path_name , 0x00 , sizeof(topic_path_name) );
		snprintf( topic_path_name , sizeof(topic_path_name) , "%s/%s" , store->store_path , dtp->d_name );
		stat( topic_path_name , & dir_stat );
		if( ! S_ISDIR(dir_stat.st_mode) )
			continue;
		
		topic = (struct MQTopic *)malloc( sizeof(struct MQTopic) ) ;
		if( topic == NULL )
		{
			ERRORLOG( "malloc MQTopic failed , errno[%d]" , errno )
			return -2;
		}
		memset( topic , 0x00 , sizeof(struct MQTopic) );
		
		strcpy( topic->topic_name , dtp->d_name );
		topic->topic_name_len = strlen(topic->topic_name) ;
		snprintf( topic->topic_path , sizeof(topic->topic_path)-1 , "%s/%s" , store->store_path , dtp->d_name );
		INIT_LIST_HEAD( & (topic->topic_file_list) );
		
		nret = LoadAllTopicDataFiles( broker , store , topic ) ;
		if( nret )
		{
			ERRORLOG( "LoadAllMessageTopicDataFiles failed , errno[%d]" , errno )
			return -3;
		}
		
		INFOLOG( "add[%.*s] to topic_name_tree" , topic->topic_name_len,topic->topic_name )
		LinkTopicNameTreeNode( store , topic );
	}
	
	closedir( dp );
	
	return 0;
}

int UnloadAllTopics( struct MQBroker *broker , struct MQStore *store )
{
	struct MQTopic		*p_topic = NULL ;
	struct MQFileMapping	*topic_file = NULL ;
	struct MQFileMapping	*next_topic_file = NULL ;
	
	p_topic = NULL ;
	for( ; ; )
	{
		p_topic = TravelTopicNameTreeNode( store , p_topic ) ;
		if( p_topic == NULL )
			break;
		
		list_for_each_entry_safe( topic_file , next_topic_file , & (p_topic->topic_file_list) , struct MQFileMapping , file_list_node )
		{
			UnloadTopicDataFile( broker , store , p_topic , topic_file );
		}
	}
	
	DestroyTopicNameTree( store );
	
	return 0;
}

